#include "file_transfer.h"
#include "socket_util.h"
#include "thread_safe_queue.h"
#include "logger.h"
#include "md5.h"
#include <fstream>
#include <sys/sendfile.h>
#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <cstring>
#include <thread>
#include <atomic>
#include <vector>

// 文件传输协议
struct FileMetadata
{
    bool is_upload;
    uint64_t file_size;
    char filename[256];
};

struct ChunkHeader
{
    uint64_t offset;
    uint64_t size;
};

struct FileChunk
{
    char data[1024 * 1024]; // 1MB
    uint64_t size;
    uint64_t offset;
};

const size_t CHUNK_SIZE = 1024 * 1024; // 1MB块大小

void upload_file(int sockfd, const std::string &local_path, const std::string &remote_path)
{
    try
    {
        // 获取文件大小
        struct stat file_stat;
        if (stat(local_path.c_str(), &file_stat))
        {
            Logger::error("Cannot get file size for " + local_path);
            return;
        }
        uint64_t file_size = file_stat.st_size;

        // 打开本地文件
        int file_fd = open(local_path.c_str(), O_RDONLY);
        if (file_fd == -1)
        {
            Logger::error("Cannot open file: " + local_path);
            return;
        }

        // 发送文件元数据
        FileMetadata metadata{};
        metadata.is_upload = true;
        metadata.file_size = file_size;
        strncpy(metadata.filename, remote_path.c_str(), sizeof(metadata.filename) - 1);
        metadata.filename[sizeof(metadata.filename) - 1] = '\0';

        if (send(sockfd, &metadata, sizeof(metadata), 0) != sizeof(metadata))
        {
            Logger::error("Failed to send metadata");
            close(file_fd);
            return;
        }

        // 接收已传输的字节数（用于断点续传）
        uint64_t existing_size = 0;
        if (recv(sockfd, &existing_size, sizeof(existing_size), 0) != sizeof(existing_size))
        {
            Logger::error("Failed to receive existing size");
            close(file_fd);
            return;
        }

        Logger::info("Uploading file: " + local_path +
                     " (" + std::to_string(existing_size) + "/" +
                     std::to_string(file_size) + " bytes)");

        // 多线程传输 - 使用生产者-消费者模式
        ThreadSafeQueue<FileChunk> chunk_queue;
        std::atomic<uint64_t> total_sent(existing_size);
        std::atomic<bool> error_occurred(false);

        // 生产者线程：读取文件块
        std::thread producer([&]
                             {
            uint64_t offset = existing_size;
            while (offset < file_size && !error_occurred) {
                FileChunk chunk;
                chunk.size = std::min(static_cast<uint64_t>(CHUNK_SIZE), file_size - offset);
                chunk.offset = offset;
                
                ssize_t bytes_read = pread(file_fd, chunk.data, chunk.size, offset);
                if (bytes_read <= 0) {
                    if (bytes_read < 0) {
                        Logger::error("File read error: " + std::string(strerror(errno)));
                        error_occurred = true;
                    }
                    break;
                }
                
                if (static_cast<uint64_t>(bytes_read) != chunk.size) {
                    Logger::warning("Partial read: expected " + 
                                   std::to_string(chunk.size) + " bytes, got " + 
                                   std::to_string(bytes_read));
                    chunk.size = bytes_read;
                }
                
                chunk_queue.push(std::move(chunk));
                offset += bytes_read;
            }
            
            chunk_queue.set_done(); });

        // 消费者线程：发送文件块
        auto consumer_func = [&]
        {
            while (!error_occurred)
            {
                auto chunk_opt = chunk_queue.pop();
                if (!chunk_opt)
                    break;

                FileChunk chunk = *chunk_opt;

                // 发送块元数据
                ChunkHeader header{chunk.offset, chunk.size};
                if (send(sockfd, &header, sizeof(header), 0) != sizeof(header))
                {
                    Logger::error("Failed to send chunk header");
                    error_occurred = true;
                    break;
                }

                // 发送块数据
                uint64_t bytes_sent = 0;
                while (bytes_sent < chunk.size && !error_occurred)
                {
                    ssize_t n = send(sockfd, chunk.data + bytes_sent, chunk.size - bytes_sent, 0);
                    if (n <= 0)
                    {
                        Logger::error("Failed to send chunk data");
                        error_occurred = true;
                        break;
                    }
                    bytes_sent += n;
                }

                if (error_occurred)
                    break;

                total_sent += chunk.size;
            }
        };

        // 创建多个消费者线程
        std::vector<std::thread> consumers;
        const int num_consumers = 4;
        for (int i = 0; i < num_consumers; ++i)
        {
            // 修复 lambda 捕获问题
            consumers.emplace_back([&chunk_queue, &sockfd, &error_occurred, &total_sent]
                                   {
                while (!error_occurred) {
                    auto chunk_opt = chunk_queue.pop();
                    if (!chunk_opt) break;
                    
                    FileChunk chunk = *chunk_opt;
                    
                    // 发送块元数据
                    ChunkHeader header{chunk.offset, chunk.size};
                    if (send(sockfd, &header, sizeof(header), 0) != sizeof(header)) {
                        Logger::error("Failed to send chunk header");
                        error_occurred = true;
                        break;
                    }
                    
                    // 发送块数据
                    uint64_t bytes_sent = 0;
                    while (bytes_sent < chunk.size && !error_occurred) {
                        ssize_t n = send(sockfd, chunk.data + bytes_sent, chunk.size - bytes_sent, 0);
                        if (n <= 0) {
                            Logger::error("Failed to send chunk data");
                            error_occurred = true;
                            break;
                        }
                        bytes_sent += n;
                    }
                    
                    if (error_occurred) break;
                    
                    total_sent += chunk.size;
                } });
        }

        producer.join();
        for (auto &t : consumers)
            t.join();

        close(file_fd);

        if (error_occurred)
        {
            Logger::error("File upload failed: " + local_path);
            return;
        }

        // 发送MD5校验
        std::string md5 = MD5Calculator::calculate(local_path);
        if (send(sockfd, md5.c_str(), md5.size(), 0) != md5.size())
        {
            Logger::error("Failed to send MD5");
        }

        Logger::info("File upload completed: " + local_path);
    }
    catch (const std::exception &e)
    {
        Logger::error("Upload error: " + std::string(e.what()));
    }
}

void download_file(int sockfd, const std::string &remote_path, const std::string &local_path)
{
    // 类似上传逻辑，略
}